package com.zyx.pulsardemo.basic.consumers;

import java.util.stream.IntStream;

import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;

/**
 * @author Yaxi.Zhang
 * @since 2021/9/30 15:23
 * desc: 异步消息处理
 * reference: https://github.com/david-streamlio/pulsar-in-action
 */
public class MessageListenerExample {

    static final String PULSAR_SERVICE_URL = "pulsar://bigdata022:6650,bigdata023:6650,bigdata024:6650";
    static final String MY_TOPIC = "persistent://manning/chapter03/example-topic";
    static final String SUBSCRIPTION = "my-shared-sub";

    public static void main(String[] args) throws PulsarClientException {

        // Pulsar客户端用来连接Pulsar
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(PULSAR_SERVICE_URL)
                .build();

        // 消费者构建工厂, 稍后将用于创建消费者实例
        ConsumerBuilder<byte[]> consumerBuilder = client.newConsumer()
                .topic(MY_TOPIC)
                .subscriptionName(SUBSCRIPTION)
                .subscriptionType(SubscriptionType.Shared)
                .messageListener((consumer, msg) -> {
                    try {
                        String consumerName = consumer.getConsumerName();
                        System.out.println("Message received: " +
                                new String(msg.getData()) + " by " + consumerName);
                        // 确认消息, 以便Broker可以删除该消息!!!
                        consumer.acknowledge(msg);
                    } catch (PulsarClientException e) {
                        // 将消息标记为重新投递
                        consumer.negativeAcknowledge(msg);
                    }
                });

        // 在主题上创建4个消费者, 每个消费者都具有相同的 MessageListener 实现
        IntStream.range(0, 4).forEach(i -> {
            String name = String.format("mq-consumer-%d", i);
            try {
                consumerBuilder
                        .consumerName(name)
                        // 将消费者连接到主题以开始接收消息
                        .subscribe();
            } catch (PulsarClientException e) {
                // TODO Auto-generated catch block
            }
        });

        // Sleep ?
    }
}